package com.diao.flink.processFunctionAPI


import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * @author: Chenzhidiao
 * @date: 2020/4/8 7:45
 * @description:
 * @version: 1.0
 */
case class SensorReading(
                          id: String,
                          timestamp: Long,
                          temperature: Double
                        )

object KeyedProcessFunctionTest {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val stream: DataStream[String] = env.socketTextStream("localhost", 7777)
    val dataStream = stream.map(data => {
      val dataArray: Array[String] = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
    }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
      override def extractTimestamp(t: SensorReading): Long = {
        t.timestamp * 1000
      }
    }) //分配时间戳和水位值

    val processStream: DataStream[String] = dataStream.keyBy(_.id).process(new TempIncreAlert())
    dataStream.print()
    processStream.print()
    env.execute()
  }

}

class TempIncreAlert() extends KeyedProcessFunction[String, SensorReading, String] {
  lazy val lastTemp: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemp", classOf[Double]))
  lazy val currentTimer: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("currentTimer", classOf[Long]))

  override def processElement(i: SensorReading, context: KeyedProcessFunction[String, SensorReading, String]#Context, collector: Collector[String]): Unit = {
    val preTemp: Double = lastTemp.value()
    lastTemp.update(i.temperature)
    val curTimerTs: Long = currentTimer.value()

    if (i.temperature > preTemp && curTimerTs == 0.0) {
      val timerTs: Long = context.timerService().currentProcessingTime() + 10000L
      context.timerService().registerProcessingTimeTimer(timerTs)
      currentTimer.update(timerTs)
    }else if(i.temperature<preTemp||preTemp==0.0){
      context.timerService().deleteProcessingTimeTimer(curTimerTs)
      currentTimer.clear()
    }
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
    out.collect(ctx.getCurrentKey + "温度连续上升")
    currentTimer.clear()
  }
}

